import oracle.jdbc.aq.AQAgent;
import oracle.jdbc.aq.AQFactory;
import oracle.jdbc.driver.OracleConnection;

import org.apache.log4j.Logger;


public class DBMS_AQADMUtil extends OracleAQ{
	
	private Logger logger = Logger.getLogger( DBMS_AQADMUtil.class);
	
	public DBMS_AQADMUtil() {}
	
	public DBMS_AQADMUtil( String user, String password, String host, String port, String sid) {
		super( user, password, host, port, sid);
	
	}
	
	public DBMS_AQADMUtil( String user, String password, String url) {
		super( user, password, url);
	
	}
	
	
	public int createQueueTable( String queueTable, String payloadType, String compatible) {
		
		try {
			this.connect();
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE( QUEUE_TABLE=>'" 
			+ this.user + "." + queueTable + "', QUEUE_PAYLOAD_TYPE=>'" + payloadType + "', COMPATIBLE=>'" + compatible + "'); END; ");
			logger.info( queueTable + " queue table is created.");
			logger.info( "queueTable = " + queueTable + ", payloadType = " + payloadType + ", compatible = " + compatible);
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
	
	}
	
	public int createQueueTable( OracleConnection conn, String queueTable, String payloadType, String compatible) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE( QUEUE_TABLE=>'" 
			+ this.user + "." + queueTable + "', QUEUE_PAYLOAD_TYPE=>'" + payloadType + "', COMPATIBLE=>'" + compatible + "'); END; ");
			logger.info( queueTable + " queue table is created.");
			logger.info( "queueTable = " + queueTable + ", payloadType = " + payloadType + ", compatible = " + compatible);
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		return 0;
	
	}
	
	public int createQueueTableTransaction ( OracleConnection conn, String queueTable, String payloadType, String compatible) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE( QUEUE_TABLE=>'" 
			+ this.user + "." + queueTable + "', QUEUE_PAYLOAD_TYPE=>'" + payloadType + "', COMPATIBLE=>'" + compatible + "', MESSAGE_GROUPING=>1); END; ");
			logger.info( queueTable + " queue table is created.");
			logger.info( "queueTable = " + queueTable + ", payloadType = " + payloadType + ", compatible = " + compatible);
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		return 0;
	
	}
	
	public int createQueueTable( OracleConnection conn, String queueTable, String payloadType, String compatible, boolean secure) {
		if( secure) {
			try {
				this.conn = conn;
				this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE( QUEUE_TABLE=>'" 
						+ this.user + "." + queueTable + "', QUEUE_PAYLOAD_TYPE=>'" + payloadType + "', COMPATIBLE=>'" + compatible + "', MULTIPLE_CONSUMERS=>true, SECURE=>true); END; ");
				logger.info( queueTable + " queue table is created.");
				logger.info( "queueTable = " + queueTable + ", payloadType = " + payloadType + ", compatible = " + compatible);
			
			} catch( Exception e) {
				logger.error( e.getMessage());
				return 1;
			
			}
		} else {
			this.createQueueTable(conn, queueTable, payloadType, compatible);
			
		}
		
		return 0;
	
	}
	
	public int dropQueueTable( String queueTable) {
		
		try {
			this.connect();
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.DROP_QUEUE_TABLE( QUEUE_TABLE=>'" + this.user
					+ "." + queueTable + "', FORCE=>TRUE); END; ");
			logger.info( queueTable + " queue table is dropped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		} finally {
			this.disconnect();
		
		}
		
		return 0;
	
	}
	
	public int dropQueueTable( OracleConnection conn, String queueTable) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.DROP_QUEUE_TABLE( QUEUE_TABLE=>'"
					+ queueTable + "', FORCE=>TRUE); END; ");
			logger.info( queueTable + " queue table is dropped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
	
	}
	
	public int createQueue( String queueName, String queueTable) {
		
		try {
			this.connect();
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE( QUEUE_NAME=>'" 
					+ queueName + "', QUEUE_TABLE=>'" + queueTable + "'); END; ");
			logger.info( queueName + " Queue is created.");
			logger.info( "queueName = " + queueName + ", queueTable = " + queueTable);
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
		
		} finally {
			this.disconnect();
			
		}
		
		return 0;

	}
	
	public int createQueue( OracleConnection conn, String queueName, String queueTable) {
		
		try {
			
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.CREATE_QUEUE( QUEUE_NAME=>'" 
					+ queueName + "', QUEUE_TABLE=>'" + queueTable + "'); END; ");
			logger.info( queueName + " Queue is created.");
			logger.info( "queueName = " + queueName + ", queueTable = " + queueTable);
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
		
	}
	
	public int dropQueue( String queueName, String queueTable) {
		
		try {
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.DROP_QUEUE( QUEUE_NAME=>'" + queueName + "'); END; ");
			logger.info( queueName + " Queue is dropped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
		
		} finally {
			this.disconnect();
			
		}
		
		return 0;

	}
	
	public int dropQueue( OracleConnection conn, String queueName, String queueTable) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.DROP_QUEUE( QUEUE_NAME=>'" + queueName + "'); END; ");
			logger.info( queueName + " Queue is dropped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
		
		}
		
		return 0;

	}
	
	public int startQueue( String queueName) {
		
		try {
			this.connect();
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.START_QUEUE('" + queueName + "'); END; ");
			logger.info( queueName + " queue start.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
		
		} finally {
			this.disconnect();
		
		}
		
		return 0;
		
	}

	public int startQueue( OracleConnection conn, String queueName) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.START_QUEUE('" + queueName + "'); END; ");
			logger.info( queueName + " queue start.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
		
	}
	
	public int stopQueue( String queueName) {
		
		try {
			this.connect();
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.STOP_QUEUE('" + queueName + "'); END; ");
			logger.info( queueName + " queue stopped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
		
		} finally {
			this.disconnect();
			
		}
		
		return 0;
	
	}
	
	public int stopQueue( OracleConnection conn, String queueName) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.STOP_QUEUE('" + queueName + "'); END; ");
			logger.info( queueName + " queue stopped.");
			
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
		
	}
	
	
	public int add_Subscriber( OracleConnection conn, String queue_Name, AQAgent subscriber) {
		
		try {
			this.conn = conn;
			this.doUpdateDatabase( "BEGIN DBMS_AQADM.ADD_SUBSCRIBER ( QUEUE_NAME=>'" + queue_Name + "', SUBSCRIBER=>" + subscriber + "); END; ");
			logger.info( "Subscriber add.");
		
		} catch( Exception e) {
			logger.error( e.getMessage());
			return 1;
			
		}
		
		return 0;
	}
	
	public static void main( String[] argv) {
		
		DBMS_AQADMUtil da = new DBMS_AQADMUtil( "strmadmin", "streams", "192.168.0.98", "1521", "bizteco");
//		DBMS_AQADMUtil da = new DBMS_AQADMUtil( "strmadmin", "streams", "192.168.0.198", "1521", "bizteco");
		
		OracleConnection conn = da.connect();
		
		
//		da.createQueueTable( conn, "OGG_TABLE", "SYS.ANYDATA", "10.0");
//		da.createQueueTable( conn, "OGG_TABLE", "SYS.ANYDATA", "10.0", true);
//		da.createQueue( conn, "OGG_QUEUE", "strmadmin.OGG_TABLE");
//		da.startQueue( conn, "OGG_QUEUE");
//		da.stopQueue( conn, "OGG_QUEUE");
//		da.dropQueue( conn , "OGG_QUEUE", "strmadmin.OGG_TABLE");
//		da.dropQueueTable( conn, "OGG_TABLE");
		
//		da.stopQueue( conn, "OGG_XML_QUEUE");
//		da.dropQueue( conn , "OGG_XML_QUEUE", "strmadmin.OGG_TABLE");
//		da.dropQueueTable( conn, "OGG_XML_TABLE");
//		da.createQueueTable( conn, "OGG_XML_TABLE", "SYS.XMLTYPE", "10.0");
//		da.createQueueTableTransaction( conn, "OGG_XML_TABLE", "SYS.XMLTYPE", "10.0");
//		da.createQueue( conn, "OGG_XML_QUEUE", "strmadmin.OGG_XML_TABLE");
		da.startQueue( conn, "OGG_XML_QUEUE");
		
		
//		da.createQueueTable( conn, "OGG_RAW_TABLE", "RAW", "10.0");
//		da.createQueueTable( conn, "OGG_TABLE", "SYS.ANYDATA", "10.0", true);
//		da.createQueue( conn, "OGG_RAW_QUEUE", "strmadmin.OGG_RAW_TABLE");
//		da.startQueue( conn, "OGG_RAW_QUEUE");
//		da.stopQueue( conn, "OGG_QUEUE");
//		da.dropQueue( conn , "OGG_QUEUE", "strmadmin.OGG_TABLE");
//		da.dropQueueTable( conn, "OGG_TABLE");
		
		
//		da.createQueueTable( conn, "Q_APPLY_TABLE", "SYS.ANYDATA", "10.0");
//		da.createQueue( conn, "Q_APPLY", "strmadmin.Q_APPLY_TABLE");
//		da.startQueue( conn, "Q_APPLY");
		
		da.disconnect();

	}
}
